package com.archgeek.kafka

import kafka.admin.TopicCommand.{CommandTopicPartition, TopicCommandOptions}
import kafka.zk.{AdminZkClient, KafkaZkClient}
import kafka.zookeeper.ZooKeeperClient
import org.apache.kafka.common.errors.TopicExistsException
import org.apache.kafka.common.security.JaasUtils
import org.apache.kafka.common.utils.Time

/**
 *
 *
 * Create by pizhihui on 2022-01-24
 */
object KafkaHelper {


  def createTopic(topicName: String, zkList: String, partitions: Integer,  replicas: Integer): Unit = {
    val zkClient = KafkaZkClient(zkList, JaasUtils.isZkSaslEnabled, 30000, 30000,
      Int.MaxValue, Time.SYSTEM)
    val adminZkClient = new AdminZkClient(zkClient)
    try {
      adminZkClient.createTopic(topicName, 2 ,2)
    } catch  {
      case e: TopicExistsException =>
        e.printStackTrace()
        throw e
    }
  }


  def main(args: Array[String]): Unit = {

    val topicName = "topic_test_zhihuipi_0124"
    val zkList = "10.9.22.23:2181,10.9.22.22:2181,10.9.22.25:2181"
    createTopic(topicName, zkList, 3, 2)

  }

}
